package com.york.tinyrpc.protoss.registry.center;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import cn.hutool.json.JSONUtil;
import com.york.tinyrpc.protoss.config.RegistryCenterConfig;
import com.york.tinyrpc.protoss.model.ServiceMetaInfo;
import com.york.tinyrpc.protoss.registry.FetchedServiceFromRegistryCenter;
import io.etcd.jetcd.*;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.etcd.jetcd.watch.WatchEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;


/**
 * Etcd注册中心
 * 是远程的注册中心，作用是管理服务提供方的接口信息，给消费方提供信息
 * 说明：
 * ETCD的Java客户端（如JetCD）默认采用异步非阻塞的编程模型。
 * 当调用kvClient.put()方法时，并不会立即执行写入操作，而是返回一个CompletableFuture对象
 * 因此下面实现过程中put之后调用的get完全是等待异步操作的执行结果而阻塞
 */
public class EtcdRegistryCenter implements RegistryCenter {

    private static final Logger log = LoggerFactory.getLogger(EtcdRegistryCenter.class);

    /**
     * Etcd-Client
     * 建立连接
     */
    private Client client;

    /**
     * Etcd-KVClient
     * 键值操作
     */
    private KV kvClient;

    /**
     * Etcd根节点，为了区分在Etcd中使用别的项目，来一个根节点
     */
    private static final String ETCD_ROOT_PATH = "/tiny-rpc/";

    /**
     * 维护本地注册服务信息（记录已注册节点信息）
     * 针对提供方
     */
    private final Set<String> localRegistryNodeKeySet = new HashSet<>();

    /**
     * 便于维护消费方从注册中心中已拉取到的服务列表缓存
     * todo 缓存的更新通过Etcd的监听机制实现
     */
    private final FetchedServiceFromRegistryCenter fetchedServiceCache = new FetchedServiceFromRegistryCenter();

    /**
     * 维护监听中的key集合,防止重复监听
     */
    private final Set<String> watchingKeySet = new ConcurrentHashSet<>();

    /**
     * 初始化
     * 就是给Etcd建立连接，并得到KV操作的客户端
     *
     * @param registryConfig 用户配置的注册信息
     */
    @Override
    public void registryCenterInit(RegistryCenterConfig registryConfig) {
        // 根据用户对注册中心的配置建立连接，设置超时时间
        client = Client.builder().endpoints(registryConfig.getAddress())
                .connectTimeout(Duration.ofMillis(registryConfig.getTimeout()))
                .build();
        kvClient = client.getKVClient();
        log.info("EtcdRegistryCenter init success...");

        // 开启定时续签注册信息
        refreshToken();
    }

    /**
     * 注册服务（服务端）
     *
     * @param serviceMetaInfo 想要注册到该注册中心的服务元信息
     */
    @Override
    public void serviceRegister(ServiceMetaInfo serviceMetaInfo) throws Exception {
        // 创建 Lease 和 KV 客户端
        Lease leaseClient = client.getLeaseClient();

        // 创建一个 30 秒的租约（过期时间）
        long leaseId = leaseClient.grant(30).get().getID();

        // 设置要存储的键值对
        String registerKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        // 构建存储到Etcd的KV
        ByteSequence key = ByteSequence.from(registerKey, StandardCharsets.UTF_8);
        ByteSequence value = ByteSequence.from(JSONUtil.toJsonStr(serviceMetaInfo), StandardCharsets.UTF_8);

        // 将键值对与租约关联起来，设置过期时间
        PutOption putOption = PutOption.builder().withLeaseId(leaseId).build();
        kvClient.put(key, value, putOption).get();
        log.info("EtcdRegistryCenter register service:{} success..., which key:{} - value:{}",
                serviceMetaInfo.getServiceName(), key, value);

        // 不仅注册到注册中心，并同步到本地的已注册记录中
        localRegistryNodeKeySet.add(registerKey);
    }

    /**
     * 注销服务（服务端）
     *
     * @param serviceMetaInfo
     */
    @Override
    public void removeServiceRegister(ServiceMetaInfo serviceMetaInfo) {
        String deleteKey = ETCD_ROOT_PATH + serviceMetaInfo.getServiceNodeKey();
        try {
            kvClient.delete(ByteSequence.from(deleteKey, StandardCharsets.UTF_8)).get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        log.warn("EtcdRegistryCenter remove service:{} success..., which key:{}",
                serviceMetaInfo.getServiceName(), deleteKey);

        // 不仅在注册中心注销，并同步注销本地的已注册记录
        localRegistryNodeKeySet.remove(deleteKey);
    }

    /**
     * 服务发现（获取某服务的所有节点，消费端）
     *
     * @param serviceKey 服务键名
     * @return
     */
    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        // 先从缓存中尝试获取服务列表，没有的话再从注册中心获取
        List<ServiceMetaInfo> fetchedServiceMetaInfos = fetchedServiceCache.readTargetCache(serviceKey);
        if (CollUtil.isNotEmpty(fetchedServiceMetaInfos)) {
            return fetchedServiceMetaInfos;
        }

        // 前缀搜索，结尾一定要加 '/'
        String searchPrefix = ETCD_ROOT_PATH + serviceKey + "/";

        try {
            // 前缀查询
            GetOption getOption = GetOption.builder().isPrefix(true).build();
            ByteSequence searchKey = ByteSequence.from(searchPrefix, StandardCharsets.UTF_8);
            List<KeyValue> keyValues = kvClient.get(searchKey, getOption).get().getKvs();

            log.info("EtcdRegistryCenter discovery service success...");

            // 解析服务信息
            List<ServiceMetaInfo> serviceMetaInfos = keyValues.stream().map(keyValue -> {
                // 得到key并监听
                String key = keyValue.getKey().toString(StandardCharsets.UTF_8);
                // 监听 key 的变化
                watchServiceStateByKey(key);
                String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                return JSONUtil.toBean(value, ServiceMetaInfo.class);
            }).collect(Collectors.toList());

            // 将获取到的服务列表信息写入拉取服务的缓存中去
            fetchedServiceCache.saveTargetServiceToCache(serviceKey, serviceMetaInfos);

            return serviceMetaInfos;
        } catch (Exception e) {
            throw new RuntimeException("获取服务列表失败", e);
        }
    }

    /**
     * 注册中心下线
     */
    @Override
    public void destroyRegistryCenter() {
        log.warn("EtcdRegistryCenter下线成功...");

        for (String key : localRegistryNodeKeySet) {
            try {
                kvClient.delete(ByteSequence.from(key, StandardCharsets.UTF_8)).get();
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }

        // 释放资源
        if (kvClient != null) {
            kvClient.close();
        }
        if (client != null) {
            client.close();
        }
    }

    /**
     * 心跳检测机制
     * 实际作用是续签，只要提供方的接口不下线，且在过期前通过续签，则重写该注册信息的过期时间
     */
    @Override
    public void refreshToken() {
        // 每10秒钟续签一次
        CronUtil.schedule(
                "*/10 * * * * *",
                (Task) () -> {
                    // 遍历本节点所有的 key
                    for (String key : localRegistryNodeKeySet) {
                        try {
                            ByteSequence kvClientKey = ByteSequence.from(key, StandardCharsets.UTF_8);
                            List<KeyValue> kvClientValues = kvClient.get(kvClientKey).get().getKvs();
                            // 说明该节点已过期（必须重启节点才能重新注册）
                            if (CollUtil.isEmpty(kvClientValues)) {
                                continue;
                            }
                            // 说明当前节点未过期，续签重刷过期时间
                            KeyValue keyValue = kvClientValues.get(0);
                            String value = keyValue.getValue().toString(StandardCharsets.UTF_8);
                            ServiceMetaInfo serviceMetaInfo = JSONUtil.toBean(value, ServiceMetaInfo.class);
                            // 重新注册 == 续签
                            serviceRegister(serviceMetaInfo);
                            log.info("服务:{} - [{} - {}] 续签成功...,",
                                    serviceMetaInfo.getServiceName(), serviceMetaInfo.getServiceNodeKey(), serviceMetaInfo.getServiceAddress());
                        } catch (Exception e) {
                            throw new RuntimeException(key + "续签失败", e);
                        }
                    }
                });

        // 支持秒级别定时任务
        CronUtil.setMatchSecond(true);
        CronUtil.start();
    }


    /**
     * 服务消费方监听【服务键名】的服务状态，动态刷新服务列表缓存
     *
     * @param serviceKey 服务键名
     */
    @Override
    public void watchServiceStateByKey(String serviceKey) {
        Watch watchClient = client.getWatchClient();
        // 之前未被监听，开启监听
        boolean newWatch = watchingKeySet.add(serviceKey);
        if (newWatch) {
            watchClient.watch(
                    ByteSequence.from(serviceKey, StandardCharsets.UTF_8),
                    response -> {
                        for (WatchEvent event : response.getEvents()) {
                            switch (event.getEventType()) {
                                // key 删除时触发,即服务注销时触发
                                case DELETE:
                                    // 清理注册服务缓存
                                    fetchedServiceCache.clearTargetCache(serviceKey);
                                    log.info("Etcd watchServiceStateByKey :{}", serviceKey);
                                    break;
                                case PUT:
                                default:
                                    break;
                            }
                        }
                    });
        }
    }

}

